package com.jxkj.mysqlbinlog;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.jxkj.entity.Student;
import com.jxkj.mapper.StudentMapper;
import com.jxkj.repository.RedisRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Set;

/**
 * 功能描述：
 *
 * @author wcx
 * @version 1.0
 */
@Component
@Order(1)
public class BinLogUtils implements ApplicationRunner {

    private final String CANAL_HOST = "182.92.89.118";
    private final Integer CANAL_PORT = 11111;
    private final String CANAL_DESTINATION = "example";
    private final String CANAL_USERNAME = "user";
    private final String CANAL_PASSWORD = "user";

    @Autowired
    private RedisRepository redisRepository;

    @Autowired
    private StudentMapper studentMapper;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        getCanalConn();
    }

    private void getCanalConn() {
        System.out.println(LocalDateTime.now());
        Set<String> keys = redisRepository.getKeys("students");
        if (keys.size() <= 0) {
            List<Student> students = studentMapper.selectList(null);
            System.out.println(students);
            redisRepository.setList("students", students);
        }
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_HOST,
                CANAL_PORT), CANAL_DESTINATION, CANAL_USERNAME, CANAL_PASSWORD);
        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    List<CanalEntry.Entry> entrys = message.getEntries();
                    for (CanalEntry.Entry entry : entrys) {
                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                            continue;
                        }

                        CanalEntry.RowChange rowChage = null;
                        try {
                            rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        } catch (Exception e) {
                            throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                                    e);
                        }

                        CanalEntry.EventType eventType = rowChage.getEventType();
                        String eventTypeStr = eventType.toString();
                        boolean flag = ("INSERT".equals(eventTypeStr) ||
                                "CREATE".equals(eventTypeStr) ||"UPDATE".equals(eventTypeStr) || "DELETE".equals(eventTypeStr));
                        System.out.println(flag);
                        if ("student".equals(entry.getHeader().getTableName()) && flag) {
                            redisRepository.del("students");
                            List<Student> students = studentMapper.selectList(null);
                            System.out.println(students);
                            redisRepository.setList("students", students);
                        }
                        System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                                entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                                entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                                eventType));
                    }
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }
}
